using System;
using System.Collections.Generic;
using System.Threading;
using ShuttleBus.AdminMessages;
using ShuttleBus.Serialization;

namespace ShuttleBus
{
    public abstract class Transport : ITransport
    {
        private readonly List<Thread> _threads = new List<Thread>();
        private bool _started;
        private volatile bool _stopped = false;
        private readonly int _maxRetries;
        private readonly int _threadCount;

        protected readonly ISerializer Serializer;
        protected readonly IEndpoint Endpoint;


        protected Transport(IEndpoint endpoint, ISerializer serializer, int threadCount, int maxRetries)
        {
            Endpoint = endpoint;
            Serializer = serializer;
            _threadCount = threadCount;
            _maxRetries = maxRetries;
            MessageArrived += m => { };

            for (int i = 0; i < _threadCount; i++)
            {
                _threads.Add(new Thread(StartListening) { Name = "ShuttleBus-" + (i + 1), IsBackground = true });
            }
        }

        public event Action<object> MessageArrived;
        public event Action<SubscriptionMessage> HandleAdministrativeMessage;

        public void Start()
        {
            if (_started)
            {
                return;
            }

            _threads.ForEach(t => t.Start(Endpoint));

            _started = true;
        }

        public void Stop()
        {
            _stopped = true;
            _threads.ForEach(t => t.Join(TimeSpan.FromSeconds(1)));
        }

        public abstract void Send(TransportMessage message, Subscription subscription);

        public void StartListening(object state)
        {
            var endpoint = (IEndpoint)state;
            using (var queue = endpoint.Initialize(QueueMode.Receive))
            {
                while (!_stopped)
                {
                    try
                    {
                        BaseMessage message;
                        if (queue.HasMessage(out message))
                        {
                            ProcessMessage(message);

                            queue.RemoveMessage(message);
                        }
                    }
                    catch (ThreadAbortException)
                    {
                        return;
                    }
                    catch
                    {
                        continue;
                    }
                }
            }
        }

        private void ProcessMessage(BaseMessage rcvdMessage)
        {
            if (rcvdMessage.RecieveCount <= _maxRetries)
            {
                var message = Serializer.Deserialize(rcvdMessage.Contents);

                if (message is SubscriptionMessage)
                {
                    var subscriptionMessage = message as SubscriptionMessage;
                    subscriptionMessage.OriginalMessage = rcvdMessage.OriginalMessage;
                    HandleAdministrativeMessage(subscriptionMessage);
                }
                else
                {
                    MessageArrived(message);
                }
            }
        }
    }

    [Serializable]
    public class AddSubscriptionMessage : SubscriptionMessage
    {

    }

    [Serializable]
    public class SubscriptionMessage : Message
    {
        public string MessageType { get; set; }
        public string Sender { get; set; }

        public object OriginalMessage { get; set; }

        public Subscription ToSubscription()
        {
            return new Subscription(new Uri(Sender), Type.GetType(MessageType));
        }
    }
}